package pubsub.impl;

import client.DtClientAware;
import pubsub.MessageListener;
import pubsub.Subscriber;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by xuan on 2018/8/16.
 */
public class DefaultSubscriber extends DtClientAware implements Subscriber {

    //订阅的消息监听
    private Map<String, List<MessageListener>> listenerMap = new ConcurrentHashMap<>();

    @Override
    public void addListener(String topic, MessageListener listener) {
        List<MessageListener> listenerList = listenerMap.putIfAbsent(topic, new ArrayList<>());
        if (null == listenerList) {
            listenerList = listenerMap.get(topic);
        }
        listenerList.add(listener);
    }

    @Override
    public void doSubscribe() {
        Set<String> topicSet = listenerMap.keySet();
        String[] topics = topicSet.toArray(new String[topicSet.size()]);

        new Thread(() -> getClient().subcribe((String topic, String message) -> {
            List<MessageListener> lList = listenerMap.get(topic);
            if (null != lList && !lList.isEmpty()) {
                lList.forEach(l -> l.onMessage(topic, message));
            }
        }, topics)).start();
    }

}
